package com.chenzhiling.study.datasource.RemoteFile

import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.sources.{BaseRelation, DataSourceRegister, RelationProvider}

/**
 * @Author: CHEN ZHI LING
 * @Date: 2021/8/26
 * @Description: 远程文件数据源
 */
class DefaultSource extends DataSourceRegister with RelationProvider {
  override def shortName(): String = "remoteFile"

  override def createRelation(sqlContext: SQLContext, parameters: Map[String, String]): BaseRelation = {
    val params: Map[String, String] = initParams(parameters)
    new RemoteFileRelation(sqlContext,params)
  }


  /**
   * 路径校验
   * @param parameters map
   * @return
   */
  private def checkPath(parameters: Map[String, String]): String = {
    parameters.getOrElse("path",
      throw new IllegalArgumentException("path must be specified."))
  }

  /**
   * 参数初始化
   * @param parameters 参数集合
   * @return
   */
  def initParams(parameters: Map[String, String]): Map[String, String] = {
    val ip: String = parameters.getOrElse("ip", throw new IllegalArgumentException("ip must be specified."))
    val port: String = parameters.getOrElse("port", throw new IllegalArgumentException("port must be specified."))
    val username: String = parameters.getOrElse("username", throw new IllegalArgumentException("username must be specified."))
    val password: String = parameters.getOrElse("password", throw new IllegalArgumentException("password must be specified."))
    val remotePath: String = checkPath(parameters)
    val params: Map[String, String] = Map[String, String](
      "ip" -> ip,
      "port" -> port,
      "username" -> username,
      "password" -> password,
      "path" -> remotePath)
    params
  }
}
